Summary

This PR improves RedisStreamBroker reliability and fairness by adopting stronger Redis Streams handling patterns: per-task-deadline reclaim, lock-free XCLAIM, listener-local prefetch backpressure, and faster handoff of buffered messages on listener close.

What changed

• Replace lock-based pending recovery with timeout-aware XCLAIM recovery.
• Reclaim messages using the task's own timeout label plus reclaim_timeout_grace, falling back to idle_timeout for messages without a timeout label.
• Deprecate and ignore unacknowledged_lock_timeout; Redis XCLAIM min_idle_time now provides the atomic claim guard.
• Add reclaim_interval so pending-message scans are throttled instead of running on every listen loop.
• Add local prefetch/backpressure tracking so xread_count caps delivered-but-unacknowledged messages for a single-stream listener.
• Deprecate additional_streams. It remains supported for compatibility, but will be removed in a future major release; use one broker and worker process per stream instead.
• Track only messages delivered by the current listener instance as protected from reclaim, so a restarted worker using the same consumer_name can still recover pending messages from its predecessor.
• On listener close, claim entries that were fetched into the broker-local buffer but not yet yielded to taskiq to an internal abandoned consumer and stamp them as very idle, so the next reclaim sweep can recover them immediately.
• Update README documentation for stream broker reclaim/backpressure/close behavior.

Why

The previous reclaim path used a single global idle_timeout plus a broker-side Redis lock. That can reclaim long tasks too early (for example, a task with a 30m timeout reclaimed at the default 10m idle_timeout) and recover short crashed tasks slowly (for example, a 5s task waiting for the default 10m idle_timeout). The lock also adds a failure point (unacknowledged_lock_timeout) and extra round-trips, while Redis XCLAIM min_idle_time already deduplicates claims atomically on the server.

additional_streams makes strict listener-level prefetch limits ambiguous: Redis applies XREADGROUP COUNT per stream, not across all streams in a multi-stream read. One broker per stream keeps consumer ownership, reclaim behavior, and prefetch capacity independent.

The close-time handoff is deliberately limited to the broker's internal buffer: messages already yielded to taskiq may be executing, so they are not abandoned immediately to avoid duplicate execution during shutdown.

Behavior notes

• Reclaim sweeps run before XREADGROUP within a listen iteration, so overdue pending messages take priority over new messages when a sweep is due.
• reclaim_interval defaults to 30000 ms; set it to 0 to scan on every iteration.
• Only messages held by the current listener instance are protected from reclaim; pending entries from a dead predecessor (even with the same consumer_name) are recoverable.
• xread_count limits delivered-but-unacknowledged messages for a single-stream broker. xread_count=None disables the limit.
• additional_streams emits DeprecationWarning when configured.
• Broker-local buffered entries are handed to an internal abandoned consumer on listener close; already-yielded entries still follow their normal ack/reclaim lifecycle.

Tests

Added coverage for:

• task-timeout based reclaim
• idle-timeout fallback reclaim
• reclaim with shared consumer_name
• xread_count backpressure
• additional_streams deprecation warning
• buffered-message handoff on listener close

Validation run locally:

uv run ruff check taskiq_redis/redis_broker.py tests/test_broker.py
uv run mypy taskiq_redis/redis_broker.py tests/test_broker.py
uv run pytest tests/test_broker.py -k "not cluster and not sentinel" -q

Result: 16 passed, 7 deselected for the single-node Redis broker tests.

Reference

The reclaim/backpressure design was informed by the Redis Streams broker implementation in sylvinus/dramatiq-redis-streams, especially its use of deadline-aware pending recovery, bounded local prefetching, and abandoned-buffer handoff on close.